package com.liuyang.concurrency.chapter12;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.stream.IntStream;

public class SimpleThreadPool extends Thread {
    private int size;
    private final int queueSize;
    private final DiscardPolicy discardPolicy;

    private final static int DEFAULT_SIZE = 10;
    private final static int DEFAULT_TASK_QUEUE_SIZE = 1000;

    private final ThreadGroup GROUP = new ThreadGroup("SIMPLE-THREAD-POOL");

    private final String THREAD_PREFIX = "SIMPLE_POOL_PREFIX-";

    private static volatile int seq = 0;


    private final static LinkedList<Runnable> TASK_QUEUE = new LinkedList<>();

    private static List<WorkerTask> THREAD_QUEUE = new ArrayList<>();

    public final static DiscardPolicy DEFAULT_DISCARD_POLICY = () -> {
        throw new DiscardException("DISCARD THIS TASK...");
    };

    private volatile boolean destroy = false;

    private int min;

    private int max;

    private int active;

    public SimpleThreadPool() {
        this(4,8,12,DEFAULT_TASK_QUEUE_SIZE,
                DEFAULT_DISCARD_POLICY);

    }

    public SimpleThreadPool(int min , int active,int max,int queueSize, DiscardPolicy discardPolicy) {
        this.min = min;
        this.max = max;
        this.active = active;
        this.queueSize = queueSize;
        this.discardPolicy = discardPolicy;
        init();
    }

    private void init() {
        for (int i = 0; i < this.min; i++) {
            createWorker();
        }
        this.size = min;
        this.start();

    }

    public void submit(Runnable runnable) {
        if (destroy == true) {
            throw new IllegalStateException("the thread pool already destroy and not allow submit task..");
        }
        synchronized (TASK_QUEUE) {
            if (TASK_QUEUE.size() >= queueSize) {
                discardPolicy.discard();
            }
            TASK_QUEUE.addLast(runnable);
            TASK_QUEUE.notifyAll();
        }
    }

    public void createWorker() {
        WorkerTask task = new WorkerTask(GROUP, THREAD_PREFIX + (seq++));
        task.start();
        THREAD_QUEUE.add(task);
    }

    public void shutdown() throws InterruptedException {
        while (!TASK_QUEUE.isEmpty()) {
            Thread.sleep(50);
        }

        int initVal = THREAD_QUEUE.size();
        while (initVal > 0) {
            for (WorkerTask workerTask : THREAD_QUEUE) {
                if (workerTask.getTaskState() == TaskState.BLOCKED) {
                    workerTask.interrupt();
                    ;
                    workerTask.close();
                    initVal--;
                } else {
                    Thread.sleep(10);
                }
            }
        }
        System.out.println("the thread pool disposed");
        this.destroy = true;
    }

    @Override
    public void run() {
        while (!destroy) {
            System.out.printf("Pool#Min:%d,Active:%d,Max:%d,Current:%d,TASK_QUEUE:%d\n",
                   this.min,this.active,this.max,this.size,this.TASK_QUEUE.size());
            try {
                Thread.sleep(1_000);
                if(TASK_QUEUE.size()>active && size < active){
                    for(int i=size; i<active;i++){
                        createWorker();
                    }
                    System.out.println("The pool incremented to active");
                    size = active;
                }else if(TASK_QUEUE.size() > max && size < max){
                    for(int i=size; i<max; i++){
                        createWorker();
                    }
                    System.out.println("The pool incremented to max");
                    size = max;
                }

                if(TASK_QUEUE.isEmpty() && size>active ){
                    synchronized (TASK_QUEUE){
                        int releaseSize = size -active;
                        Iterator<WorkerTask> iterator = THREAD_QUEUE.iterator();
                        while(iterator.hasNext()){
                            if(releaseSize <= 0){
                                break;
                            }
                            WorkerTask task = iterator.next();
                            if(TaskState.FREE == task.taskState){
                                task.close();
                                task.interrupt();
                                iterator.remove();
                                releaseSize--;
                            }

                        }
                        size = active;
                    }
                }

            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    public boolean isDestroy() {
        return this.destroy;
    }

    public int getSize() {
        return size;
    }

    public int getQueueSize() {
        return queueSize;
    }

    public int getMin() {
        return min;
    }

    public int getMax() {
        return max;
    }

    public int getActive() {
        return active;
    }

    private enum TaskState {
        FREE, RUNNING, BLOCKED, DEAD
    }


    public static interface DiscardPolicy {
        void discard() throws DiscardException;
    }


    private static class WorkerTask extends Thread {
        private volatile TaskState taskState = TaskState.FREE;

        public TaskState getTaskState() {
            return this.taskState;
        }

        public void close() {
            this.taskState = TaskState.DEAD;
        }

        public WorkerTask(ThreadGroup group, String name) {
            super(group, name);
        }

        public void run() {
            OUTER:
            while (this.taskState != TaskState.DEAD) {
                Runnable runnable;
                synchronized (TASK_QUEUE) {
                    while (TASK_QUEUE.isEmpty()) {
                        try {
                            taskState = TaskState.BLOCKED;
                            TASK_QUEUE.wait();
                        } catch (InterruptedException e) {
                            break OUTER;
                        }
                    }
                    runnable = TASK_QUEUE.removeFirst();
                }

                if (runnable != null) {
                    taskState = TaskState.RUNNING;
                    runnable.run();
                    taskState = TaskState.FREE;
                }
            }
        }

    }

    public static void main(String[] args) throws InterruptedException {
        SimpleThreadPool pool = new SimpleThreadPool();
        for (int t = 0; t <= 40; t++) {
            int finalT = t;
            new Thread(() -> pool.submit(() -> {
                System.out.println("the task " + finalT + " by service by" + Thread.currentThread() + "...start");
                try {
                    Thread.sleep(3_000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("the task " + finalT + " by service by" + Thread.currentThread() + "...end");

            })).start();
        }

        Thread.sleep(20_000);
//        pool.start();
        pool.shutdown();
    }

    public static class DiscardException extends RuntimeException {
        public DiscardException(String message) {
            super(message);
        }
    }
}
